package com.ibm.spatiotemp.job.processor;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.Collection;
import java.util.LinkedList;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

import com.ibm.spatiotemp.datamodel.datatype.Constraints;
import com.ibm.spatiotemp.datamodel.datatype.TemporalLineString;
import com.ibm.spatiotemp.datamodel.datatype.request.TSBuildIdxRequest;
import com.ibm.spatiotemp.datamodel.datatype.timeseries.GPSTimeseries;
import com.ibm.spatiotemp.job.JobID;
import com.ibm.spatiotemp.processor.splitter.LinestringSplitter;
import com.ibm.spatiotemp.processor.splitter.LinestringSplitterFactory;

import edu.umn.cs.spatialHadoop.OperationsParams;
import edu.umn.cs.spatialHadoop.core.Point;
import edu.umn.cs.spatialHadoop.core.Rectangle;
import edu.umn.cs.spatialHadoop.operations.Repartition;

public class TSIndexBuildingProcessor {

	public static int RunningMode = Constraints.DATASRC_MODE_MOCK;

	public static Logger logger = Logger
			.getLogger(TSIndexBuildingProcessor.class);

	private GPSTimeseries timesreies;
	
	/*for support 1st layer index building*/
	public TSIndexBuildingProcessor(){}
	public TSIndexBuildingProcessor(GPSTimeseries ts){
		this.timesreies = ts;
	}
	/* output meta for 1st layer index building */
	public Collection<TemporalLineString> buildRtreeIndex(JobID id,
			TSBuildIdxRequest request) throws IOException, InterruptedException {
		Collection<TemporalLineString> rawlinestr = splitTimeseriesDatatoLineString(timesreies);
		Collection<TemporalLineString> outputlinestr = retrieveLineStringMeta(rawlinestr);

		String linestringpath = request
				.getLineStringBuildingPath(id.getJobid());
		try {
			writeLinestringset2HDFS(rawlinestr, linestringpath);
		} catch (Exception e) {
		}
		rawlinestr = null;

		String rtreepath = request.getRtreePath();
		// linestringpath = "test_linestr/data_00001";
		buildRtree(linestringpath, rtreepath);
		return outputlinestr;
	}

	private Collection<TemporalLineString> splitTimeseriesDatatoLineString(GPSTimeseries ts) {
		// throw new RuntimeException("not implemented");
//		LinkedList<TemporalLineString> linestrset = new LinkedList<TemporalLineString>();
		LinestringSplitter splitter =
				LinestringSplitterFactory.getSplitterInstance(RunningMode);
		
		Collection<TemporalLineString> linestrset  =
				splitter.splitTimeseriesDatatoLineString(ts);
		return linestrset;
	}

	public static TemporalLineString transformTemporalLineStringToMBR(
			TemporalLineString target) throws Exception {
		Rectangle rec = target.getMBR();
		long starts = target.base_timestamp;
		long endts = (long) target.timestamp_vector[target.point_cnt - 1]
				+ starts;
		Point startpos = new Point(rec.x1, rec.y1);
		Point endpos = new Point(rec.x2, rec.y2);
		TemporalLineString result = new TemporalLineString();
		result.ts_id = target.ts_id;
		result.insertPoint(starts, startpos);
		result.insertPoint(endts, endpos);
		return result;

	}

	private Collection<TemporalLineString> retrieveLineStringMeta(
			Collection<TemporalLineString> rawdata) {
		Collection<TemporalLineString> resultset = new LinkedList<TemporalLineString>();
		for (TemporalLineString curlinestr : rawdata) {
			try {
				resultset.add(transformTemporalLineStringToMBR(curlinestr));
			} catch (Exception e) {
				logger.error("linestring transformantion failed");
			}
		}
		return resultset;
	}

	public boolean writeLinestringset2HDFS(
			Collection<TemporalLineString> target, String hdfspath)
			throws Exception {
		Path inFile = new Path(hdfspath);
		FileSystem inFs = inFile.getFileSystem(new Configuration());
		FSDataOutputStream fout = inFs.create(new Path(hdfspath));
		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fout));
		for (TemporalLineString curline : target) {
			writer.write(curline.toString());
			writer.newLine();
		}
		writer.flush();
		writer.close();
		return true;
	}

	public boolean buildRtree(String linestrset_hdfspath, String rtree_outpath)
			throws IOException, InterruptedException {

		/*
		 * inputFile outputFile path may be input paramter in a new function
		 * given later
		 */
		OperationsParams params = new OperationsParams();
		params.initialize(linestrset_hdfspath, rtree_outpath);
		Path outputFile = params.getOutputPath();
		Path inputFile = params.getInputPath();
		FileSystem fs = inputFile.getFileSystem(new Configuration());

		// LinkedList<Path> resultpath = new LinkedList<Path>();
		if (fs.isDirectory(inputFile)) {
			// FileStatus[] status = fs.listStatus(inputFile);
			// for (FileStatus file : status) {
			// if (!file.getPath().getName().startsWith("data")) {
			// continue;
			// }
			// resultpath.add(file.getPath());
			// }
			logger.error("not a file, is a folder");
			return false;
		}

		/*
		 * command line example: test_linestr sindex:rtree test_rtree
		 * shape:templinestr
		 */
		params.set("sindex", "rtree");
		params.set("shape", "templinestr");
		params.setBoolean("local", true);

		/* just for test overwrite cabability */
		params.setBoolean("overwrite", false);

		// The spatial index to use
		long t1 = System.currentTimeMillis();
		Repartition.repartition(inputFile, outputFile, params);
		long t2 = System.currentTimeMillis();
		logger.info("Total indexing time in millis " + (t2 - t1));

		return true;
	}
}
